package com.example.rocketmq.producer.config;

import com.alibaba.fastjson2.JSON;
import com.example.rocketmq.producer.dto.UserInfoDTO;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

@Configuration
public class FunctionConfig {

    // 定义了一个java.util.function.Supplier类型的producer函数，它负责生成消息并返回
    @Bean
    public Supplier<String> producerStr() {
        return () -> "Hello from Producer!";
    }

    @Bean
    public Supplier<String> producerUser() {
        return () -> {
            UserInfoDTO userInfo = UserInfoDTO.builder()
                    .name("张三")
                    .age(18)
                    .gender(true)
                    .birthdate(LocalDate.now())
                    .phoneNumber("123-456-7890")
                    .email("example@example.com")
                    .website("https://example.com")
                    .creditCardNumber("1234-5678-9012-3456")
                    .price(BigDecimal.valueOf(10.0))
                    .build();
            return JSON.toJSONString(userInfo);
        };
    }

    // 消费者，一个应用可以即是生产者，又是消费者
    @Bean
    public Consumer<String> consumerStr() {
        return message -> {
            System.out.println("Consumer Str: " + message);
        };
    }

    @Bean
    public Consumer<String> consumerUser() {
        return message -> {
            String data = JSON.toJSONString(message);
            System.out.println("Consumer User: " + data);
        };
    }


    // 对于函数式 Bean 来说，如果是 Function 类型，则需要定义两个泛型，第一个泛型为输入类型，第二个泛型为输出类型

    // 批处理消费者
    // 当使用支持批处理侦听器的MessageChannelBinder时，并且为消费者绑定启用了该功能，
    // 您可以将spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode设置为true以启用将整个消息批次传递给函数中的List。
    @Bean
    public Function<List<String>, String> findFirstPerson() {
        return List::getFirst;
    }

    // 批处理生产者
    // 参考官方 https://docs.springjava.cn/spring-cloud-stream/reference/spring-cloud-stream/producing-and-consuming-messages.html
    // 您也可以在生产者端使用批处理的概念，方法是返回一个消息集合，这实际上提供了和批处理消费相反的效果，其中集合中的每条消息都将由绑定程序单独发送
    @Bean
    public Function<String, List<Message<String>>> batch() {
        return p -> {
            List<Message<String>> list = new ArrayList<>();
            list.add(MessageBuilder.withPayload(p + ":1").build());
            list.add(MessageBuilder.withPayload(p + ":2").build());
            list.add(MessageBuilder.withPayload(p + ":3").build());
            list.add(MessageBuilder.withPayload(p + ":4").build());
            return list;
        };
    }


    // 为下面的函数式组合提供初始数据源，不然下面两个函数根本无法开始工作。因为他们都依赖一个 Message<String> 类型的初始数据源。
    @Bean
    public Supplier<Message<String>> func() {
        return () -> MessageBuilder.withPayload("Hello World 2024 from func!").build();
    }

    // 函数式组合，参考 https://docs.springjava.cn/spring-cloud-stream/reference/spring-cloud-stream/producing-and-consuming-messages.html#polling-configuration-properties
    @Bean
    public Function<Message<String>, Message<String>> enrich() {
        return message -> {
            // Assert.isTrue(!message.getHeaders().containsKey("foo"), "Should NOT contain 'foo' header");
            return MessageBuilder.fromMessage(message).setHeader("foo", "bar").build();
        };
    }

    @Bean
    public Function<Message<String>, Message<String>> echo() {
        return message -> {
            // Assert.isTrue(message.getHeaders().containsKey("foo"), "Should contain 'foo' header");
            System.out.println("Incoming message " + message);
            return message;
        };
    }


    // 使用 StreamBridge 的通道拦截器
    // 由于 StreamBridge 使用 MessageChannel 建立输出绑定，因此您可以在通过 StreamBridge 发送数据时激活通道拦截器。应用程序决定对 StreamBridge 应用哪些通道拦截器。
    // Spring Cloud Stream 不会将检测到的所有通道拦截器注入到 StreamBridge 中，除非它们用 @GlobalChannelInterceptor(patterns = "*") 注解。
    @Bean
    @GlobalChannelInterceptor(patterns = "*")
    public ChannelInterceptor customInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                System.out.println("Pre send: " + message);
                return message;
            }
        };
    }

    // 如果您不喜欢上面的全局方法，而希望为每个绑定提供一个专门的拦截器，则可以执行以下操作。
    // 式语法：通常使用 Ant 风格的路径匹配模式，如 * 表示任意字符，? 表示单个字符，** 表示任意层级的目录等
    @Bean
    @GlobalChannelInterceptor(patterns = "producer-*")
    public ChannelInterceptor fooInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                return MessageBuilder.withPayload("hello" + ":1").build();
            }
        };
    }

    @Bean
    @GlobalChannelInterceptor(patterns = "bar-*")
    public ChannelInterceptor barInterceptor() {
        return new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                return MessageBuilder.withPayload("hello" + ":1").build();
            }
        };
    }



}
